package socketmvc.core.packets;

import java.util.Set;
import java.util.concurrent.*;
import java.util.concurrent.locks.LockSupport;
import java.util.function.Consumer;

/**
 * 可以控制结果的Future,可主动set结果给future
 * @author xueyanjun
 */
public class ControllableFuture<P> implements Future<P> {
    // 所有等待响应包的线程
    protected final Set<Thread> threads;

    // 所有等待响应包的异步回调函数
    protected final Set<Consumer<P> > asyncCalls;

    // 响应包
    protected volatile P pkt;

    // 当前future的创建时间,单位是毫秒值
    protected final long createAt;

    // 当前future的状态 0-waiting 1-done 2-cancel
    protected volatile byte state;

    public ControllableFuture() {
        this(ConcurrentHashMap.newKeySet(), ConcurrentHashMap.newKeySet());
    }

    public ControllableFuture(Set<Thread> threadContainer) {
        this(threadContainer, ConcurrentHashMap.newKeySet());
    }

    public ControllableFuture(Set<Thread> threadContainer,Set<Consumer<P>> asyncCallContainer) {
        this.createAt = System.currentTimeMillis();
        this.state = 0;
        this.threads = threadContainer;
        this.asyncCalls = asyncCallContainer;
    }


    public long getCreateAt() {
        return createAt;
    }

    @Override
    public boolean cancel(boolean mayInterruptIfRunning) {
        this.state = 2;
        this._notify();
        return true;
    }


    @Override
    public boolean isCancelled() {
        return this.state == 2;
    }


    @Override
    public boolean isDone() {
        return this.state == 1;
    }


    @Override
    public P get() {
        return this.getOrNull();
    }

    @Override
    public P get(long timeout, TimeUnit unit) throws TimeoutException {
        return this.getOrThrow(timeout,unit);
    }

    public P getOrNull(long timeout, TimeUnit unit){
        if (this.pkt == null){
            threads.add(Thread.currentThread());
            LockSupport.parkNanos(unit.toNanos(timeout));
        }
        return this.pkt;
    }

    public P getOrNull() {
        if (this.pkt == null){
            threads.add(Thread.currentThread());
            LockSupport.park();
        }
        return pkt;
    }

    public P getOrThrow(long timeout, TimeUnit unit) throws TimeoutException,CancellationException {
        if (this.pkt == null){
            threads.add(Thread.currentThread());
            LockSupport.parkNanos(unit.toNanos(timeout));
            if (this.pkt == null){
                if (this.isCancelled()){
                    throw new CancellationException();
                }
                throw new TimeoutException("obtain packet timeout " + unit.toSeconds(timeout) + "s");
            }
        }
        return this.pkt;
    }

    public void then(Consumer<P> callback){
        asyncCalls.add(callback);
    }

    public void set(P pkt){
        this.state = 1;
        this.pkt = pkt;
        try {
            for (Consumer<P> asyncCall : asyncCalls) {
                asyncCall.accept(this.pkt);
            }
        }finally {
            this._notify();
        }
    }

    private void _notify(){
        for (Thread thread : this.threads) {
            LockSupport.unpark(thread);
        }
    }
}
